//package com.cheng.im.service.massage.mq;
//
//import com.cheng.im.common.bean.msg.ProtoMsgV1;
//import com.cheng.im.common.constants.MQConstants;
//import com.cheng.im.service.massage.service.impl.MessageSyncService;
//import com.cheng.im.service.massage.service.impl.P2PMessageService;
//import com.google.protobuf.InvalidProtocolBufferException;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//import javax.annotation.Resource;
//import java.util.List;
//
//@Service
//@Slf4j
//public class ChatOperateListen {
//
//    @Value("${rocketmq.name-server}")
//    private String namesrvAddr;
//    @Resource
//    P2PMessageService p2PMessageService;
//    @Resource
//    MessageSyncService messageSyncService;
//
//    private final String consumerGroup = "P2P_MSG_CONSUMER_GROUP";
//
//    private DefaultMQPushConsumer defaultMQPushConsumer;
//
//    @PostConstruct
//    public void startConsumers() {
//
//        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
//        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
//        try {
//            defaultMQPushConsumer.subscribe(MQConstants.IM_2_MSG_SERVICE, "*");
//        } catch (MQClientException e) {
//            log.error("===============消息监听订阅主题出错================");
//            throw new RuntimeException(e);
//        }
//
//        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                // 处理消息
//                for (MessageExt msg : msgs) {
//                    try {
//                        log.info("消费消息");
//                        byte[] body = msg.getBody();
//                        ProtoMsgV1.Message pkg = ProtoMsgV1.Message.parseFrom(body);
//                        ProtoMsgV1.HeadType headType = pkg.getType();
//
//                        // 下面判断各种消息类型，业务处理、消息接收ack、消息已读ack、消息撤回等等
//                        if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_REQUEST)) {
//                            p2PMessageService.process(pkg);
//                        } else if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_RECEIVE_ACK)) {
//                            // 接收端ack
//                            messageSyncService.receiveMark(pkg);
//                        } else if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_READ)) {
//                            // 用户已读消息
//                            messageSyncService.readMark(pkg);
//                        }
//                    } catch (InvalidProtocolBufferException e) {
//                        log.error("MQ消息监听的消息出错了");
//                        // 重试消费
//                        e.printStackTrace();
//                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                    }
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//    }
//
//    @PreDestroy
//    public void destroy(){
//        defaultMQPushConsumer.shutdown();
//    }
//}
